package multisource.kafkademo;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.source.TransactionSource;
import sampleexec.SimepleDetail;
import sampleexec.SimepleEntitySink;
import sampleexec.SimpleEntity;
import spendreport.FraudDetector;

import java.util.Properties;

public class KfkTest {
    public static  void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "selton.cn:9092");
        properties.setProperty("group.id", "test");
        DataStream<String> transactions = env
                .addSource(new FlinkKafkaConsumer<>("test", new SimpleStringSchema(), properties))
                .name("transactions");

        DataStream<SimpleEntity> alerts = transactions
//                .keyBy(Transaction::getAccountId)
                .process(new SimepleDetail())
                .name("fraud-detector");

        alerts
                .addSink(new SimepleEntitySink())
                .name("send-alerts");

        env.execute("Fraud Detection");

    }
}